package com.lili.collection.restful.plugins;

import cn.hutool.core.text.CharSequenceUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONUtil;
import com.lili.collection.core.context.BatchRecordContext;
import com.lili.collection.core.context.StageContext;
import com.lili.collection.core.domain.RunnerMsgTransfer;
import com.lili.collection.core.enums.DataFormat;
import com.lili.collection.core.error.OutputRunnerException;
import com.lili.collection.core.error.StageRecordException;
import com.lili.collection.core.plugins.OutputPlugin;
import com.lili.collection.core.record.Record;
import com.lili.collection.core.runner.result.ErrOutputData;
import com.lili.collection.core.runner.result.ErrorDetail;
import com.lili.collection.core.runner.result.RecordInfo;
import com.lili.collection.core.runner.result.RunnerResult;
import com.lili.collection.core.utils.HandlerUtil;
import com.lili.collection.core.utils.NoticeUtil;
import com.lili.collection.mq.domain.ByteMessageWrapper;
import com.lili.collection.mq.utils.JsonConvertUtil;
import com.lili.collection.restful.base.IInterfaceRequest;
import com.lili.collection.restful.base.InterfaceRequestFactory;
import com.lili.collection.restful.config.InterfaceConfig;
import com.lili.collection.restful.config.RestfulOutputConfig;
import com.lili.collection.restful.domain.ResponseCheckException;
import com.lili.collection.restful.enums.InterfaceType;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.List;

/**
 *描述: 接口输出插件
 *@date: 2023/12/18  14:36
 *@author: lili
 */
@Slf4j
public class RestfulOutputPlugin implements OutputPlugin<RestfulOutputConfig> {
    private final static int MAX_MESSAGE_SIZE = 1024 * 1024 * 4;

    private RestfulOutputConfig config;
    private Throwable tempThrowable = null;
    private int executeNum;
    @Getter
    private RunnerResult runnerResult;
    @Getter
    private RunnerMsgTransfer transfer;
    private int maxMessageSize;
    /**
     * 接口配置
     */
    private InterfaceConfig interfaceConfig;

    /**
     * 接口请求对象
     */
    private IInterfaceRequest request;

    @Override
    public void init(RestfulOutputConfig config) {
        try {
            this.config = config;
            if(this.config==null){
                throw new IllegalArgumentException("接口采集输出配置为空!");
            }
            this.createInterfaceConfig();
            this.request = InterfaceRequestFactory.getInfRequest(InterfaceType.codeOf(this.interfaceConfig.getInterfaceType()));
            if(this.config.getDataFormat()==null){
                throw new IllegalArgumentException("接口输入输出数据格式定义不能为空！");
            }
            this.maxMessageSize = MAX_MESSAGE_SIZE;
        } catch (Exception e) {
            log.error("创建接口生产者客户端异常!", e);
            throw new OutputRunnerException(e);
        }
    }



    private void createInterfaceConfig() {
        String interfaceConfig = this.config.getInterfaceConfig();
        if(StringUtils.isBlank(interfaceConfig)){
            throw new IllegalArgumentException("接口参数配置为空!");
        }
        this.interfaceConfig = JSONUtil.toBean(interfaceConfig, InterfaceConfig.class);
        this.interfaceConfig.init();
    }

    @Override
    public void setRunnerResult(RunnerResult outputRunnerResult) {
        this.runnerResult = outputRunnerResult;
    }

    @Override
    public void setRunnerTransfer(RunnerMsgTransfer transfer) {
        this.transfer = transfer;
    }

    @Override
    public synchronized void execute(StageContext stageContext) {
        BatchRecordContext recordContext = (BatchRecordContext) stageContext;
        RecordInfo recordInfo = new RecordInfo();
        this.executeNum++;
        RunnerResult<String> outputRunnerResult = this.getRunnerResult();
        List<Record> output = recordContext.getOutput();
        try {
            try {
                if (this.getTransfer() != null && this.getTransfer().isRunError()) {
                    throw new IllegalStateException("接口异步运行错误!");
                }
                if (output.size() == 0) {
                    log.info("-->所有数据写入完成!");
                    return;
                }
                if (config.getDataFormat() == DataFormat.ORIGIN || config.getDataFormat()==DataFormat.ORIGIN_CONVERT) {
                    for (Record record : output) {
                        this.sendData(String.valueOf(record.getField().getValue()));
                        recordInfo.incrementSuccessCount();
                    }
                } else {
                    List<ByteMessageWrapper> wrapperList = JsonConvertUtil.convertRecordToJsonByteWrapper(output, recordInfo, this.maxMessageSize, false);
                    for (ByteMessageWrapper byteMessageWrapper : wrapperList) {
                        //消息推送
                        try {
                            HandlerUtil.sendProcessedData(recordInfo,config, byteMessageWrapper.getOrginJsonData(), (data) -> {
                                this.sendData(data);
                                return true;
                            });
                            if(config.getDataFormat() != DataFormat.OBJECT){
                                recordInfo.addSuccessCount(byteMessageWrapper.getDataRowSize());
                            }
                        } catch (Throwable throwable) {
                            if(config.getDataFormat() == DataFormat.OBJECT){
                                recordInfo.addErrorCount(byteMessageWrapper.getDataRowSize()-recordInfo.getSuccessRecordCount());
                            }else {
                                recordInfo.addErrorCount(byteMessageWrapper.getDataRowSize());
                            }
                            String errorMsg = "接口消息发送失败!";
                            recordInfo.addErrorValueDetail(new ErrorDetail("JSON", byteMessageWrapper.getOrginJsonData(), errorMsg, byteMessageWrapper.getBytes().length));
                            throw throwable;
                        }
                    }
                }
                //转换消息
                log.info("-->第{}批次数据写入完成\r\n-->写入量：{}", recordContext.getIndex(), output.size());
            } catch (Throwable throwable) {
                log.error("推送接口错误!", throwable);
                recordInfo.getErrOutputData().putErrorInfo(throwable.toString());
                outputRunnerResult.setSuccessed(false);
                throw throwable;
            }
        } catch (Throwable throwable) {
            tempThrowable = throwable;
            RunnerMsgTransfer transfer = this.getTransfer();
            if (transfer != null) {
                transfer.setRunError(true);
                transfer.interruptRunner();
            }
            recordContext.setStageRecordState(StageContext.StageRecordState.ALL_ERROR);
            recordContext.setException(new StageRecordException(throwable));
            outputRunnerResult.setSuccessed(false);
            //错误日志收集
            ErrOutputData errOutputData = recordInfo.getErrOutputData();
            errOutputData.setBatchIndex(recordContext.getIndex());
            errOutputData.setErrorMessage(throwable.toString());
            errOutputData.setFirstBatchError(executeNum == 1 ? true : false);
            errOutputData.setType("BATCH-OUTPUT-RESTFUL");
            log.info("-->第{}批次数据写入时发生异常\r\n-->原因：{}",
                    recordContext.getIndex(),
                    throwable.getLocalizedMessage(), throwable);
        }
        if (tempThrowable != null || !recordInfo.isSuccessed()) {
            outputRunnerResult.mergeErrorRecord(recordInfo.getErrOutputData());
        }
        this.addResultCount(recordInfo, outputRunnerResult);
        recordContext.clearOutput();
    }

    private void sendData(String data) throws Exception{
        if(StringUtils.isBlank(data)){
            throw new IllegalArgumentException("发送接口数据不能为空！");
        }
        NoticeUtil.sendDataAndSaveNotice(config,data,(sendValue)->{
            Object response= request.pushOutputData(this.interfaceConfig, sendValue);
            this.checkResult(response);
            return true;
        });
    }

    /**
     * 校验结果数据
     * @param data 返回数据
     */
    private void checkResult(Object data) {
        if ("Y".equals(interfaceConfig.getReturnCodeFlag())) {
            if (ObjectUtil.isEmpty(data)) {
                throw new ResponseCheckException("返回结果为空！");
            }
            if (CharSequenceUtil.hasBlank(interfaceConfig.getReturnCode(), interfaceConfig.getReturnSuccessFlag(), interfaceConfig.getReturnMessageCode())) {
                throw new ResponseCheckException("返回码字段、返回码成功标识、返回码说明字段均不能为空！");
            }
            if (!JSONUtil.isJsonObj((String) data)) {
                log.error("响应数据非JSON格式！请求参数：{}，返回结果：{}", JSONUtil.toJsonStr(interfaceConfig), data);
                throw new ResponseCheckException("响应数据非JSON格式！");
            }
            cn.hutool.json.JSON json = JSONUtil.parse(data);
            //不是成功的响应
            if (!interfaceConfig.getReturnSuccessFlag().equals(json.getByPath(interfaceConfig.getReturnCode()))) {
                String[] messageCodeList = interfaceConfig.getReturnMessageCode().split(",");
                String message = null;
                for (String s : messageCodeList) {
                    message = (String) json.getByPath(s);
                    if (CharSequenceUtil.isNotBlank(message)) {
                        break;
                    }
                }
                log.error("校验接口返回参数异常！请求参数：{}，返回结果：{}", JSONUtil.toJsonStr(interfaceConfig), data);
                throw new ResponseCheckException(CharSequenceUtil.isNotBlank(message) ? message : "根据返回说明字段【" + interfaceConfig.getReturnMessageCode() + "】未获取到返回提示消息");
            }
        }
    }

    private void addResultCount(RecordInfo recordInfo, RunnerResult<String> outputRunnerResult) {
        outputRunnerResult.addSuccessCount(recordInfo.getSuccessRecordCount());
        outputRunnerResult.addTotalCount(recordInfo.getTotalRecordCount());
        outputRunnerResult.addErrorCount(recordInfo.getErrorRecordCount());
        if (outputRunnerResult.isSuccessed()) {
            outputRunnerResult.setSuccessed(recordInfo.isSuccessed());
        }
    }

    @Override
    public void dispose() {
    }
}